package transport;

import com.google.protobuf.ByteString;
import hdfs.TransPort;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.codec.protobuf.ProtobufEncoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import transport.handler.ClientHandler;

import static common.Utils.PACKET_SIZE;

public class ClientTransPorter {
    private NioEventLoopGroup eventExecutors;
    private Channel channel;
    private final int blockSize;
    private static final Logger LOG = LoggerFactory.getLogger(ClientTransPorter.class);

    /**
     * 客户端写入数据工具类
     * @param size 传输数据长度
     */
    public ClientTransPorter(int size){
        this.blockSize = size;
    }

    /**
     *  返回该size需要多少个packet传输
     * @param size 传输大小
     * @return packet个数
     */
    private int packetCount(int size){
        return (size+ PACKET_SIZE-1) /PACKET_SIZE;
    }

    public void connect(String IP,int PORT) throws InterruptedException {
        eventExecutors = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(eventExecutors);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel socketChannel) throws Exception {
                ChannelPipeline pipeline = socketChannel.pipeline();
//                pipeline.addLast("decoder",new ProtobufDecoder(TransPort.WritePacketResponse.getDefaultInstance()));
                pipeline.addLast(new ProtobufVarint32FrameDecoder());
                pipeline.addLast(new ProtobufDecoder(TransPort.WritePacketRequest.getDefaultInstance()));
                pipeline.addLast(new ProtobufVarint32LengthFieldPrepender());
                pipeline.addLast(new ProtobufEncoder());
                pipeline.addLast(new ClientHandler(packetCount(blockSize),eventExecutors));
            }
        });
        ChannelFuture f = bootstrap.connect(IP, PORT).sync();
        channel = f.channel();
        LOG.info("connect {} success",PORT);
    }

    /**
     * 向管道写入数据，长度为 new ClientTransPorter时指定的size
     * @param data 数据
     * @param blockNum 块个数
     * @return 是否成功
     * @throws InterruptedException
     */
    public boolean writeBlock(byte[] data,int blockNum) throws InterruptedException {
        int offset = 0;
        ChannelFuture f = null;
        LOG.info("streamer start write block:{},packets num:{}",blockNum,packetCount(blockSize));
        while (offset<blockSize){
            TransPort.WritePacketRequest.Builder req = TransPort.WritePacketRequest.newBuilder();
            int size = Math.min(blockSize-offset,PACKET_SIZE);
            req.setData(ByteString.copyFrom(data,offset,size));
            req.setSize(size);
            req.setBlockNum(blockNum);
            offset+=size;
            f = channel.writeAndFlush(req.build());
        }
        assert f != null;
        f.channel().closeFuture().sync();
        LOG.info("streamer write block successfully");
        return true;
    }

    public void shutDown(){
        eventExecutors.shutdownGracefully();
    }
}
